
/**
* 上海中赢金融信息服务有限公司
* Copyright (c) 2017, chinazyjr All Rights Reserved.
*/
package com.sys.kafka.core.consumer.listener;

import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import com.sys.kafka.core.model.ConsumerMessage;

/**
 * <b>Description：</b> TODO <br/>
 * <b>ClassName：</b> ConsumerWorker <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2017年7月13日 下午3:27:19 <br/>
 * <b>@version: </b> <br/>
 */
public class ConsumerWorker implements Runnable {

    private ConsumerRecord<String, String> consumerRecord;

    public ConsumerWorker(ConsumerRecord<String, String> record) {
        this.consumerRecord = record;
    }

    public void run() {
        
        System.out.println("ConsumerWorker=====" +consumerRecord);
//        ConsumerMessageBO consumerMessageBO = JSONObject.parseObject(consumerRecord.value(), ConsumerMessageBO.class);
        
        ConsumerMessage consumerMessage = new ConsumerMessage();
        
        consumerMessage.setTopic(consumerRecord.topic());
        consumerMessage.setValue(consumerRecord.value());
        consumerMessage.setOffset(consumerRecord.offset());
        consumerMessage.setPartition(consumerRecord.partition());
        
       Map<String, MessageListener> listeners= MessageContainer.getAll();
       
       for (Map.Entry<String, MessageListener> entry : listeners.entrySet()) {
           String topic = entry.getKey();
           MessageListener listnener= entry.getValue();
           if(consumerRecord.topic().equals(topic)){
               listnener.onMessage(consumerMessage);
           }
           
           
           System.out.println("key= " + entry.getKey() + " and value= " + entry.getValue());
       }
        
//        for (MessageContainer messageContainer : PropertyFactory.kafkaConfig4Consumer.getMessageContainers()) {
//            if (consumerRecord.topic().equals(messageContainer.getTopic())) {
//                System.out.println("ConsumerWorker---topic----" + consumerRecord.topic() );
//                messageContainer.getMessageHandle().onMessage(consumerMessage);
//            }
//        }
    }
}
